package org.example.mq.brokerserver.datacenter;

import org.example.mq.brokerserver.core.Binding;
import org.example.mq.brokerserver.core.Exchange;
import org.example.mq.brokerserver.core.MQqueue;
import org.example.mq.brokerserver.core.Message;
import org.example.mq.common.MqException;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

//用来管理内存中的数据
//值得注意的是这里要关注多线程安全问题
public class MemoryDataCenter {
    //key为exchangeName，value为exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    //key为queueName，value为MQqueue对象
    private ConcurrentHashMap<String, MQqueue> queueMap = new ConcurrentHashMap<>();
    //key为exchangeName，value为Map对象（key为queueName，value为Binding对象
    private ConcurrentHashMap<String,ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    //key为MessageUuid，value为Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    //key为queueName，value为Message的链表
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    //key为queueName，value为Map（key为messageUuid，value为Message对象）
    private ConcurrentHashMap<String,ConcurrentHashMap<String,Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();
    //封装交换机操作
    public void insertExchange(Exchange exchange){
        exchangeMap.put(exchange.getName(),exchange);
    }
    public Exchange getExchange(String exchangeName){
        return exchangeMap.get(exchangeName);
    }
    public void deleteExchange(String exchangeName){
        exchangeMap.remove(exchangeName);
        System.out.println("[MemoryDataCenter] 交换机删除成功! exchangeName=" + exchangeName+exchangeMap.get(exchangeName));
    }
    //封装队列操作
    public void insertQueue(MQqueue queue){
        queueMap.put(queue.getName(),queue);
    }
    public MQqueue getMQqueue(String queueName){
        return queueMap.get(queueName);
    }
    public void deleteMQqueue(String queueName){
        queueMap.remove(queueName);
    }
    //封装绑定操作
    public void insertBinding(Binding binding) throws MqException{
        //先检查一下是否存在，不存在就创建一个
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),k -> new ConcurrentHashMap<>());
        synchronized (bindingMap){
            //再使用queueName来查一下，如果已经存在就抛出异常
            if(bindingMap.get(binding.getQueueName())!=null){
                throw new MqException("[MemoryDataCenter] 绑定已经存在！exchangeName="+
                        binding.getExchangeName()+", queueName="+binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(),binding);
        }
        System.out.println("[MemoryDataCenter] 新绑定添加成功！exchangeName="+binding.getExchangeName()+
                ", queueName="+binding.getQueueName());
    }
    //获取绑定
    //两个版本：1.通过exchangeName和queueName来获取唯一一个
    //2.通过exchange来获取所有的binding
    public Binding getBinding(String exchangeName,String queueName){
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(exchangeName);
        if(bindingMap==null) return null;
        return bindingMap.get(queueName);
    }
    public ConcurrentHashMap<String,Binding> getBindings(String exchangeName){
        return bindingsMap.get(exchangeName);
    }
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if(bindingMap==null){
            throw new MqException("[MemoryDataCenter] 绑定已经不存在！"+"exchangeName="+binding.getExchangeName());
        }
        bindingMap.remove(binding.getQueueName());

    }
    public void addMessage(Message message){
        messageMap.put(message.getUuid(),message);
    }
    public void deleteMessage(Message message){
        messageMap.remove(message.getUuid());
    }
    public Message getMessage(String messageId){
        return messageMap.get(messageId);
    }
    //发送值定消息到值定队列
    public void sendMessage(MQqueue queue,Message  message){
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(),k->new LinkedList<>());
        synchronized (messages){
            messages.add(message);
        }
        addMessage(message);
    }
    public Message pollMessage(String queueName){
        LinkedList<Message> messages= queueMessageMap.get(queueName);
        if(messages==null){
            return null;
        }
        synchronized (messages){
            if(messages.isEmpty()){
                return null;
            }
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] Id为"+currentMessage.getUuid()+"的消息从队列中取出了！");
            return currentMessage;
        }
    }
    public void addMessageWaitAck(String queueName,Message message){
       Map<String,Message> messageMap = queueMessageWaitAckMap.computeIfAbsent(queueName,k->new ConcurrentHashMap<>());
       messageMap.put(message.getUuid(),message);
    }
    //获取指定的未确认的消息
    public Message getMessageWaitAck(String queueName,String messageUuid){
        ConcurrentHashMap<String,Message> messageMap = queueMessageWaitAckMap.get(queueName);
        if(messageMap==null)
            return null;
        return messageMap.get(messageUuid);
    }
    //将硬盘上的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        //恢复交换机数据
        List<Exchange> exchanges  = diskDataCenter.selectAllExchange();
        for(Exchange exchange:exchanges){
            exchangeMap.put(exchange.getName(),exchange);
        }
        //恢复所有队列数据
        List<MQqueue> queues = diskDataCenter.selectAllQueue();
        for(MQqueue queue:queues){
            queueMap.put(queue.getName(),queue);
        }
        //恢复绑定
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for(Binding binding:bindings){
            ConcurrentHashMap<String,Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),k->new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(),binding);
        }
        //恢复消息
        for(MQqueue queue:queues){
            LinkedList<Message> messages=diskDataCenter.loadAllMessageFromQueue(queue);
            queueMessageMap.put(queue.getName(),messages);
            for(Message message:messages)
                messageMap.put(message.getUuid(),message);
        }
    }

    public void deleteMessageWaitAck(String queueName, String uuid) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            return;
        }
        messageHashMap.remove(uuid);
    }

    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if(messages == null){
            return 0;
        }
        synchronized (messages){
            return messages.size();
        }
    }
}













